package controllers

import (
	"context"
	"github.com/astaxie/beego"
	"github.com/astaxie/beego/orm"
	"io/ioutil"
	"os/exec"
	"strings"
	"sync"
	"time"
)

type JobEntry struct {
	jobDtl     *TblJobDtl
	cancelCtx  context.Context
	cancelFunc context.CancelFunc
	cmdContext *exec.Cmd
}

type JobMap struct {
	lock     sync.RWMutex
	jobEntry map[int64]*JobEntry
}

var (
	runScript string
	jobMap    *JobMap
)

func init() {
	runScript = beego.AppConfig.String("runScript")

	jobMap = new(JobMap)
	jobMap.jobEntry = make(map[int64]*JobEntry, 10)
}

func (job *JobMap) get(id int64) (*JobEntry, bool) {
	job.lock.RLock()
	defer job.lock.RUnlock()

	entry, ok := job.jobEntry[id]
	return entry, ok
}

func (job *JobMap) set(id int64, entry *JobEntry) {
	job.lock.Lock()
	defer job.lock.Unlock()

	job.jobEntry[id] = entry
}

func (job *JobMap) delete(id int64) {
	job.lock.Lock()
	defer job.lock.Unlock()

	delete(job.jobEntry, id)
}

func (jobDtl *TblJobDtl) updateSucc(o orm.Ormer, cols ...string) {
	jobDtl.JobStatus = JOB_STATUS_SUCC
	jobDtl.JobStatusDesc = "success"
	jobDtl.UpdateTime = time.Now().Format("2006-01-02 15:04:05")
	cols = append(cols, "job_status", "job_status_desc", "update_time")
	if _, err := o.Update(jobDtl, cols...); err != nil {
		beego.Error(err)
	}
	o.Commit()
}

func (jobDtl *TblJobDtl) updateFail(o orm.Ormer, statusDesc string, cols ...string) {
	jobDtl.JobStatus = JOB_STATUS_FAIL
	jobDtl.JobStatusDesc = statusDesc
	jobDtl.UpdateTime = time.Now().Format("2006-01-02 15:04:05")
	cols = append(cols, "job_status", "job_status_desc", "update_time")
	if _, err := o.Update(jobDtl, cols...); err != nil {
		beego.Error(err)
	}
	o.Commit()
}

func (jobDtl *TblJobDtl) updateStatus(o orm.Ormer, cols ...string) {
	jobDtl.JobStatus = JOB_STATUS_RUN
	jobDtl.UpdateTime = time.Now().Format("2006-01-02 15:04:05")
	cols = append(cols, "max_failures", "job_status", "update_time")
	if _, err := o.Update(jobDtl, cols...); err != nil {
		beego.Error(err)
	}
	o.Commit()
}

func (jobDtl *TblJobDtl) updateMaxFailures(o orm.Ormer, cols ...string) {
	jobDtl.UpdateTime = time.Now().Format("2006-01-02 15:04:05")
	cols = append(cols, "max_failures", "update_time")
	if _, err := o.Update(jobDtl, cols...); err != nil {
		beego.Error(err)
	}
	o.Commit()
}

func (jobDtl *TblJobDtl) init(o orm.Ormer) (dtl *TblJobDtl) {
	var (
		num  int64
		rows []*TblJobDtl
	)

	jobDtl.JobDate = time.Now().Format("2006-01-02")

	num, err := o.QueryTable("tbl_job_dtl").
		Filter("job_date", jobDtl.JobDate).
		Filter("job_id", jobDtl.JobId).
		OrderBy("-create_time").
		All(&rows)
	if err != nil && err != orm.ErrNoRows {
		beego.Error(err)
		return
	}

	for _, row := range rows {
		timeout := false

		createTime, err := time.ParseInLocation("2006-01-02 15:04:05 +0000 UTC", row.CreateTime, time.Local)
		if err != nil {
			beego.Error(err)
			return
		}

		if time.Now().Sub(createTime) > time.Second*time.Duration(row.Timeout) {
			timeout = true
		}

		if row.ExecTimes == JOB_EXEC_ONCE {
			switch row.JobStatus {
			case JOB_STATUS_SUCC:
				jobMap.delete(row.Id)
				return nil
			case JOB_STATUS_INIT:
				jobMap.delete(row.Id)
				return row
			default:
				if entry, ok := jobMap.get(row.Id); ok {
					if row.JobStatus == JOB_STATUS_FAIL {
						jobMap.delete(row.Id)
						row.MaxFailures -= 1
						row.updateMaxFailures(o)
						beego.Warn(row.JobId, row.JobName, "failed last time, try again.")
						return row
					}

					if timeout {
						setMonitor(row)
						entry.cancelFunc()
						entry.cmdContext.Process.Kill()
						jobMap.delete(row.Id)

						row.MaxFailures -= 1
						row.updateFail(o, "job detect timeout", "max_failures")
						beego.Warn(row.JobId, row.JobName, "timeout last time, kill and try again.")
						return row
					}
				} else {
					if row.JobStatus == JOB_STATUS_FAIL {
						row.MaxFailures -= 1
						row.updateMaxFailures(o)
						beego.Warn(row.JobId, row.JobName, "failed last time, try again.")
						return row
					}

					if timeout {
						setMonitor(row)
						row.MaxFailures -= 1
						row.updateFail(o, "job detect timeout", "max_failures")
						beego.Warn(row.JobId, row.JobName, "failed last time, try again.")
						return row
					} else {
						beego.Warn(row.JobId, row.JobName, "server restart before ???")
					}
				}
			}
		}
	}

	if num == 0 || jobDtl.ExecTimes == JOB_EXEC_MORE {
		jobDtl.CreateTime = time.Now().Format("2006-01-02 15:04:05")
		jobDtl.UpdateTime = jobDtl.CreateTime
		jobDtl.Id, err = o.Insert(jobDtl)
		if err != nil {
			beego.Error(err)
			return nil
		}
		o.Commit()
		return jobDtl
	}
	return nil
}

func (jobDtl *TblJobDtl) depend(o orm.Ormer) bool {
	var (
		row TblJobDtl
	)

	if jobDtl.MaxFailures < 1 {
		beego.Warn(jobDtl.JobId, jobDtl.JobName, "the task reached the maximum number of retries.")
		setMonitor(jobDtl)
		return false
	}

	if len(jobDtl.DependJobId) > 0 {
		jobIds := strings.Split(jobDtl.DependJobId, ",")

		for _, jobId := range jobIds {
			if err := o.QueryTable("tbl_job_dtl").
				Filter("job_date", time.Now().Format("2006-01-02")).
				Filter("job_id", jobId).
				Filter("job_status", JOB_STATUS_SUCC).One(&row); err != nil {
				if err == orm.ErrNoRows {
					beego.Warn(jobDtl.JobId, jobDtl.JobName, "dependency task", jobId, "not complete.")
				} else {
					beego.Error(err)
				}
				return false
			}
		}
	}

	if len(jobDtl.DependCustom) > 0 {
		return isDependCustomOk(jobDtl.DependCustom)
	}

	return true
}

func (jobDtl *TblJobDtl) Run() {
	beego.Debug(jobDtl.JobId, jobDtl.JobName, jobDtl.Command, "run...")

	if jobDtl.ExecDay == JOB_EXEC_WORKDAY && !isWorkDay() {
		beego.Debug(jobDtl.JobId, jobDtl.JobName, jobDtl.Command, "today is not workday!")
		return
	}

	if jobDtl.ExecDay == JOB_EXEC_HOLIDAY && isWorkDay() {
		beego.Debug(jobDtl.JobId, jobDtl.JobName, jobDtl.Command, "today is not holiday!")
		return
	}

	o := orm.NewOrm()

	newJob := jobDtl.init(o)
	if newJob == nil {
		beego.Debug(jobDtl.JobId, jobDtl.JobName, jobDtl.Command, "ignore this time.")
		return
	}

	if !newJob.depend(o) {
		beego.Debug(jobDtl.JobId, jobDtl.JobName, jobDtl.Command, "ignore this time.")
		return
	}

	entry := new(JobEntry)
	entry.jobDtl = newJob
	entry.cancelCtx, entry.cancelFunc = context.WithCancel(context.TODO())

	newJob.JobStatus = JOB_STATUS_RUN
	newJob.updateStatus(o)

	command := make([]string, 2)
	command = strings.Split(newJob.Command, " ")

	if len(runScript) > 0 {
		entry.cmdContext = exec.CommandContext(entry.cancelCtx, runScript, command...)
	} else {
		entry.cmdContext = exec.CommandContext(entry.cancelCtx, command[0], command[1:]...)
	}

	cmdStderr, err := entry.cmdContext.StderrPipe()
	if err != nil {
		beego.Error(err)
		newJob.updateFail(o, err.Error())
		return
	}
	defer cmdStderr.Close()

	jobMap.set(newJob.Id, entry)

	if err := entry.cmdContext.Start(); err != nil {
		beego.Error(err)
		newJob.updateFail(o, err.Error())
		return
	}

	output, _ := ioutil.ReadAll(cmdStderr)

	if err := entry.cmdContext.Wait(); err != nil {
		if len(output) > 0 {
			statusDesc := strings.Replace(string(output), "\r\n", "->", -1)
			statusDesc = strings.Replace(statusDesc, "\n", "->", -1)
			newJob.updateFail(o, statusDesc)
		} else {
			newJob.updateFail(o, err.Error())
		}
		return
	}

	if entry.cmdContext.ProcessState.Success() {
		newJob.updateSucc(o)
		return
	}

	return
}
